package com;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import core.Config;
import interfaces.ICommunicator;

/**
 * Classe responsável por criar a comunicação entre os componentes
 * 
 *
 */
public class Communicator extends Thread implements ICommunicator {

	String defaultIP;
	int DEFAULT_PORT = Config.DEFAULT_PORT;
	
	public static final byte NODE_STARTED = Config.NODE_STARTED;
	public static final byte NEW_WORKER = Config.NEW_WORKER;
	public static final byte BOSS_CONTACT = Config.BOSS_CONTACT;
	public static final byte NEW_JOB = Config.NEW_JOB;
	public static final byte NODE_STATUS = Config.NODE_STATUS;
	public static final byte JOB_RESULT = Config.JOB_RESULT;

	SocketChannel clientChannel = null;
	ServerSocketChannel serverChannel = null;
	InetSocketAddress channelAddress = null;
	BlockingQueue<ByteBuffer> incoming = new LinkedBlockingQueue<ByteBuffer>();
	public static Map<String, SocketChannel> clientSocketList;
	private boolean active;

	public Communicator(String defaultIP) {
		int aux = DEFAULT_PORT;
		this.defaultIP = defaultIP;
		boolean created = false;
		clientSocketList = new LinkedHashMap<String, SocketChannel>();

		while (!created) {
			try {
				serverChannel = ServerSocketChannel.open();
				channelAddress = new InetSocketAddress(this.defaultIP, aux);
			} catch (IOException e) {
				e.printStackTrace();
				System.exit(1);
			}
			try {
				serverChannel.socket().bind(channelAddress);
				created = true;
			} catch (IOException e) {
				aux++;
			}
		}
		this.active = true;
		this.start();
	}

	public Communicator() {
		
	}
	
	@Override
	public void connectServer(String hostDescription) {
		try {
			String vet[] = hostDescription.split(":");
			String hostname = vet[0];
			int port = Integer.parseInt(vet[1].trim());
			clientChannel = SocketChannel.open(new InetSocketAddress(hostname, port));
		} catch (IOException e) {
			e.printStackTrace();
			System.exit(1);
		}
		rodaListener();
	}

	public void run() {
		while (this.active) {
			try {
				clientChannel = serverChannel.accept();
				clientSocketList.put(clientRemoteChannelDesc(), clientChannel);
				rodaListener();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public class Listener extends Thread {
		private boolean listening;
		SocketChannel clientChannel;
		int nBytes;

		public Listener(SocketChannel clientChannel) {
			this.clientChannel = clientChannel;
			this.listening = true;
		}

		public void run() {
			try {
				while (listening) {
					ByteBuffer header = bytesRead(6);
					if (header == null) {
						clientChannel.close();
					} else {
						header.getShort(); // Msg_Type
						int size = header.getInt();
						header.rewind();

						ByteBuffer body = bytesRead(size - 6);
						ByteBuffer msg = ByteBuffer.allocate(size);
						msg.put(header).put(body).flip();

						try {
							incoming.put(msg);
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
				System.exit(1);
			}
		}

		public ByteBuffer bytesRead(int bytes) throws IOException {
			ByteBuffer msg = ByteBuffer.allocate(bytes);
			int readBytes = 0;
			while (readBytes < bytes) {
				int readCount = clientChannel.read(msg);
				if (readCount == -1) {
					clientChannel.close();
				}
				readBytes += readCount;
			}
			msg.flip();
			return msg;
		}
	}

	private void rodaListener() {
		Listener l = new Listener(clientChannel);
		l.start();
	}

	@Override
	public SocketChannel getSocket() {
		// TODO Auto-generated method stub
		return clientChannel;
	}

	@Override
	public String serverChannelDescription() {
		// TODO Auto-generated method stub
		String hostAddress = channelAddress.getAddress().getHostAddress();
		String portAddress = Integer.toString(channelAddress.getPort());
		return hostAddress + ":" + portAddress;
	}

	@Override
	public String clientLocalChannelDesc() {
		// TODO Auto-generated method stub
		String hostAddress = clientChannel.socket().getLocalAddress().getHostAddress();
		String portAddress = Integer.toString(clientChannel.socket().getLocalPort());
		return hostAddress + ":" + portAddress;
	}

	@Override
	public String clientRemoteChannelDesc() {
		// TODO Auto-generated method stub
		String hostAddress = clientChannel.socket().getInetAddress().getHostAddress();
		String portAddress = Integer.toString(clientChannel.socket().getPort());
		return hostAddress + ":" + portAddress;
	}

	@Override
	public ByteBuffer receiveMessages() {
		// TODO Auto-generated method stub
		try {
			return incoming.take();
		} catch (Exception e) {
			e.printStackTrace();
			System.exit(1);
			return null;
		}
	}

	@Override
	public void sendMsgNodeStatus(SocketChannel channel, String workerId, float capacity, int workload, long updateDate) {
		// TODO Auto-generated method stub
		// TIPO_MSG | TAMANHO | WORKER_ID | WORKLOAD | CAPACITY | UPDATEDATE
		// SHORT INT | INT | STRING |  FLOAT  | INT | LONG

		byte[] workerId_b = null;
		try {
			workerId_b = workerId.getBytes("ISO-8859-1");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
			System.exit(1);
		}
		int size = 2 + 4 + workerId_b.length + 1 + 4 + 4 + 8;
		ByteBuffer writeBuffer = ByteBuffer.allocateDirect(size);
		writeBuffer.putShort(NODE_STATUS);
		writeBuffer.putInt(size);
		writeBuffer.put(workerId_b);
		writeBuffer.put((byte) 0);
		writeBuffer.putFloat(capacity);
		writeBuffer.putInt(workload);
		writeBuffer.putLong(updateDate);
		writeBuffer.rewind();
		channelWrite(channel, writeBuffer);
	}

	@Override
	public void sendMsgBossContact(SocketChannel channel, String workerId) {
		// TODO Auto-generated method stub
		// TIPO_MSG | TAMANHO | WORKER_ID
		// SHORT INT | INT | STRING

		byte[] workerId_b = null;
		try {
			workerId_b = workerId.getBytes("ISO-8859-1");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
			System.exit(1);
		}
		int size = 2 + 4 + workerId_b.length + 1;
		ByteBuffer writeBuffer = ByteBuffer.allocateDirect(size);
		writeBuffer.putShort(BOSS_CONTACT);
		writeBuffer.putInt(size);
		writeBuffer.put(workerId_b);
		writeBuffer.put((byte) 0);
		writeBuffer.rewind();
		channelWrite(channel, writeBuffer);
	}

	@Override
	public void sendMsgJobResult(SocketChannel channel, String workerId, String jobId, byte[] resultDescription) {
		// TODO Auto-generated method stub
		// TIPO_MSG | TAMANHO | JOB_ID | DESC
		// SHORT INT | INT | STRING | VETOR DE BYTES
		byte[] jobId_b = null;
		byte[] workerId_b = null;
		try {
			workerId_b = workerId.getBytes("ISO-8859-1");
			jobId_b = jobId.getBytes("ISO-8859-1");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
			System.exit(1);
		}
		int size = 2 + 4 + workerId_b.length + 1 + jobId_b.length + 1
				+ resultDescription.length;
		ByteBuffer writeBuffer = ByteBuffer.allocateDirect(size);
		writeBuffer.putShort(JOB_RESULT);
		writeBuffer.putInt(size);
		writeBuffer.put(workerId_b);
		writeBuffer.put((byte) 0);
		writeBuffer.put(jobId_b);
		writeBuffer.put((byte) 0);
		writeBuffer.put(resultDescription);
		writeBuffer.rewind();
		channelWrite(channel, writeBuffer);
	}

	@Override
	public void sendMsgNewJob(SocketChannel channel, String jobId, byte[] jobDescription) {
		// TODO Auto-generated method stub
		// TIPO_MSG | TAMANHO | JOB_ID | DESC
		// SHORT INT | INT | STRING | VETOR DE BYTES

		byte[] jobId_b = null;
		try {
			jobId_b = jobId.getBytes("ISO-8859-1");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
			System.exit(1);
		}
		int size = 2 + 4 + jobId_b.length + 1 + jobDescription.length;
		ByteBuffer writeBuffer = ByteBuffer.allocateDirect(size);
		writeBuffer.putShort(NEW_JOB);
		writeBuffer.putInt(size);
		writeBuffer.put(jobId_b);
		writeBuffer.put((byte) 0);
		writeBuffer.put(jobDescription);
		writeBuffer.rewind();
		channelWrite(channel, writeBuffer);
	}

	@Override
	public void sendMsgNewWorker(SocketChannel channel, String nodeId, String nodeChannelDescription) {
		// TODO Auto-generated method stub
		// TIPO_MSG | TAMANHO | WORKER_ID | PORT
		// SHORT INT | INT | STRING | STRING

		byte[] nodeId_b = null;
		byte[] nodeChannelDescription_b = null;
		try {
			nodeId_b = nodeId.getBytes("ISO-8859-1");
			nodeChannelDescription_b = nodeChannelDescription
					.getBytes("ISO-8859-1");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
			System.exit(1);
		}
		int size = 2 + 4 + nodeId.length() + 1
				+ nodeChannelDescription.length() + 1;
		ByteBuffer writeBuffer = ByteBuffer.allocateDirect(size);
		writeBuffer.putShort(NEW_WORKER);
		writeBuffer.putInt(size);
		writeBuffer.put(nodeId_b);
		writeBuffer.put((byte) 0);
		writeBuffer.put(nodeChannelDescription_b);
		writeBuffer.put((byte) 0);
		writeBuffer.rewind();
		channelWrite(channel, writeBuffer);
	}

	@Override
	public void sendMsgNodeStarted(SocketChannel channel, String nodeId, String nodeChannelDescription) {
		// TODO Auto-generated method stub
		// TIPO_MSG | TAMANHO | NODE_ID | PORT
		// SHORT INT | INT | STRING | STRING

		byte[] nodeId_b = null;
		byte[] nodeChannelDescription_b = null;
		try {
			nodeId_b = nodeId.getBytes("ISO-8859-1");
			nodeChannelDescription_b = nodeChannelDescription
					.getBytes("ISO-8859-1");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
			System.exit(1);
		}
		int size = 2 + 4 + nodeId_b.length + 1
				+ nodeChannelDescription_b.length + 1;
		ByteBuffer writeBuffer = ByteBuffer.allocateDirect(size);
		writeBuffer.putShort(NODE_STARTED);
		writeBuffer.putInt(size);
		writeBuffer.put(nodeId_b);
		writeBuffer.put((byte) 0);
		writeBuffer.put(nodeChannelDescription_b);
		writeBuffer.put((byte) 0);
		writeBuffer.rewind();
		channelWrite(channel, writeBuffer);
	}

	@Override
	public void channelWrite(SocketChannel channel, ByteBuffer writeBuffer) {
		// TODO Auto-generated method stub
		long nbytes = 0;
		long toWrite = writeBuffer.remaining();
		try {
			while (nbytes != toWrite) {
				nbytes += channel.write(writeBuffer);

				try {
					Thread.sleep(200);
				} catch (InterruptedException e) {
				}
			}
		} catch (ClosedChannelException cce) {
		} catch (Exception e) {
		}
		writeBuffer.rewind();
	}

	public static String readString(ByteBuffer in) {
		// TODO Auto-generated method stub
		try {
			Charset charset = Charset.forName("ISO-8859-1");
			CharsetDecoder decoder = charset.newDecoder();
			ByteBuffer buf = ByteBuffer.allocate(in.capacity());
			byte b;
			b = in.get();
			while (b != 0) {
				buf.put(b);
				b = in.get();
			}
			buf.rewind();
			String s = decoder.decode(buf).toString();
			return s;
		} catch (Exception e) {
			e.printStackTrace();
			System.exit(1);
			return null;
		}
	}
}
